package com.chukun.multi.producerconsumer.workstealing;

import com.chukun.multi.producerconsumer.ProductChannel;
import com.chukun.multi.twophase.AbstractTerminatableThread;
import com.chukun.multi.twophase.TerminationToken;

import java.util.Random;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * 工作窃取算法示例
 * @author chukun
 */
public class WorkStealingExample {

    private final WorkStealingEnabledChannel<String> channel;
    private final TerminationToken terminationToken = new TerminationToken();

    /**
     * 初始化方法
     */
    public WorkStealingExample() {
        int nCpu = Runtime.getRuntime().availableProcessors();
        int consumerCount = nCpu/2+1;
        BlockingDeque<String>[] managedQueues = new LinkedBlockingDeque[consumerCount];
        // 该通道实例对应了多个队列实例managedQueues
        channel = new WorkStealingChannel<>(managedQueues);
        ProductConsumer[] consumers = new ProductConsumer[consumerCount];
        for (int i =0;i<consumerCount;i++) {
            managedQueues[i] = new LinkedBlockingDeque<>();
            consumers[i] = new ProductConsumer(terminationToken,managedQueues[i]);
        }
        for (int i = 0;i<nCpu;i++) {
            new ProductProducer().start();
        }
        for (int i =0;i<consumerCount;i++) {
            consumers[i].start();
        }
    }

    public void doSomething() {

    }

    /**
     * 生产者线程
     */
    private class ProductProducer extends AbstractTerminatableThread {

        private int i = 0;

        @Override
        protected void doRun() throws Exception {
            channel.put(String.valueOf(i++));
            terminationToken.reservations.incrementAndGet();
        }
    }


    /**
     * 消费者线程
     */
    private class ProductConsumer extends AbstractTerminatableThread{

        private final BlockingDeque<String> workQueue;

        public ProductConsumer(TerminationToken terminationToken,BlockingDeque<String> workQueue) {
            super(terminationToken);
            this.workQueue = workQueue;
        }

        @Override
        protected void doRun() throws Exception {
            /**
             * 实现 {@link WorkStealingEnabledChannel}接口实现的 {@link WorkStealingEnabledChannel#take(BlockingDeque)} 的方法
             * 实现了工作窃取算法
             */
            String product = channel.take(workQueue);
            System.out.println("process product: "+product);
            try{
                Thread.sleep(new Random().nextInt(1000));
            }catch (InterruptedException e) {
                ;
            }finally {
                terminationToken.reservations.decrementAndGet();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        WorkStealingExample wse = new WorkStealingExample();

        wse.doSomething();

        Thread.sleep(3500);
    }
}
